package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.service.PointsSortGroupService;
import com.navinfo.tripanalysis.util.CommonUtils;
import lombok.Data;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * @author 沈东生
 */
@Data
public class PointsSortGroupServiceImpl1 implements PointsSortGroupService, Serializable {
    private static final Logger logger = LoggerFactory.getLogger(PointsSortGroupServiceImpl1.class);
    /**
     * 是否删除重复数据：gpstime相同，则保留第一个gpstime相同的点
     */
    private boolean removeDuplicate = false;

    @Override
    public JavaPairRDD<Long, List<Point>> sortAndGroup(JavaRDD<Point> pointsRdd) {
        //tid->point 键值对形式
        JavaPairRDD<Long, Point> pairByTid = pointsRdd.mapPartitionsToPair((PairFlatMapFunction<Iterator<Point>, Long, Point>) pointIterator ->
            new Iterator<Tuple2<Long, Point>>(){
                @Override
                public boolean hasNext() {
                    return pointIterator.hasNext();
                }
                @Override
                public Tuple2<Long, Point> next() {
                    Point p = pointIterator.next();
                    return new Tuple2<>(p.getTid(), p);
                }
            });

        //tid->List<Point> 这种形式
        JavaPairRDD<Long, List<Point>> rtn = pairByTid.combineByKey(
            (Function<Point, List<Point>>) point -> Stream.of(point).collect(Collectors.toList()),
            (Function2<List<Point>, Point, List<Point>>) (points, p) ->{
                points.add(p);
                return points;
            } ,
            (Function2<List<Point>, List<Point>, List<Point>>) (p1, p2) ->{
                p1.addAll(p2);
                return p1;
            }
         )
        //每个分区中按gpsTime 排序
        .mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, List<Point>>>, Long, List<Point>>) tuple2Iterator ->
            new Iterator<Tuple2<Long, List<Point>>>() {
                @Override
                public boolean hasNext() {
                    return tuple2Iterator.hasNext();
                }
                @Override
                public Tuple2<Long, List<Point>> next() {
                    Tuple2<Long, List<Point>> tuple = tuple2Iterator.next();

                    List<Point> list=tuple._2;
                    List<Point> sortedPoints=null;
                    if(!CommonUtils.isCollectionEmpty(list)){
                        List<Point> sortedPoints0 = StreamSupport.stream(list.spliterator(), false)
                                .sorted(Point.COMPARATOR).collect(Collectors.toList());

                        if(removeDuplicate){
                            //按gpstime去重
                            sortedPoints = new ArrayList<>(sortedPoints0.size());
                            Long lastGpsTime =null;
                            for(Point p:sortedPoints0){
                                if(lastGpsTime ==null){
                                    sortedPoints.add(p);
                                    lastGpsTime = p.getGpsTime();
                                }else{
                                    if(!lastGpsTime.equals(p.getGpsTime())){
                                        sortedPoints.add(p);
                                        lastGpsTime = p.getGpsTime();
                                    }
                                }
                            }
                        }else{
                            sortedPoints=sortedPoints0;
                        }
                    }
                    return new Tuple2<>(tuple._1, sortedPoints);
                }
            });

        return rtn;
    }
}
